package com.avoid.easymqtt.common.algorithm;

import com.avoid.easymqtt.common.SystemClock;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicReferenceArray;
import java.util.concurrent.locks.ReentrantLock;

/**
 * <p>
 * 统计指标的基本数据结构。
 * </P>
 * <p>
 * LeapArray使用滑动窗口算法对数据进行计数。每个桶盖的{@link #windowLengthInMs}为1ms时间跨度，</br>
 * 总时间跨度为{@link #intervalInMs}，因此桶的总数量为：sampleCount=intervalInMs/windowLengthInMs。
 * </p>
 *
 * @author avoid@fiture.com
 * @date 2021/11/19
 * @motto Life is so short,do something to make yourself happy,such as coding
 */
public abstract class LeapArray<T> {

    /**
     * 单口滑动时间窗口的大小
     */
    protected int windowLengthInMs;
    /**
     * 滑动时间窗口个数
     */
    protected int sampleCount;
    /**
     * 统计时间区间(毫秒)
     */
    protected int intervalInMs;

    /**
     * 统计时间区间(秒)
     */
    private double intervalInSecond;

    /**
     * size
     */
    protected final AtomicReferenceArray<WindowWrap<T>> array;

    /**
     * 仅当当前bucket弃用时，才使用conditional lock。
     */
    private final ReentrantLock updateLock = new ReentrantLock();

    /**
     * 总桶计数为：{@code sampleCount=intervalInMs/windowLengthInMs}。
     *
     * @param sampleCount 滑动窗口的桶计数
     * @param intervalInMs 此{@link LeapArray}的总时间间隔（毫秒）
     */
    public LeapArray(int sampleCount, int intervalInMs) {

        this.windowLengthInMs = intervalInMs / sampleCount;
        this.intervalInMs = intervalInMs;
        this.intervalInSecond = intervalInMs / 1000.0;
        this.sampleCount = sampleCount;

        this.array = new AtomicReferenceArray<>(sampleCount);
    }

    /**
     * 为bucket创建一个新的统计值。
     *
     * @param timeMillis 当前时间（毫秒）
     * @return the new empty bucket
     */
    public abstract T newEmptyBucket(long timeMillis);

    /**
     * 将给bucket重置为提供的开始时间，并重置该值。
     *
     * @param startTime bucket的开始时间（以毫秒为单位）
     * @param windowWrap 当前 bucket
     * @return
     */
    protected abstract WindowWrap<T> resetWindowTo(WindowWrap<T> windowWrap, long startTime);

    /**
     * 根据当前时间计算出当前时间属于那个滑动窗口的数组下标
     */
    private int calculateTimeIdx(long timeMillis) {
        // 利用除法取整原则，保证了一秒内的所有时间搓得到的timeId是相等的
        long timeId = timeMillis / windowLengthInMs;
        // 利用求余运算原则，保证一秒内获取到的桶的下标位是一致的
        return (int)(timeId % array.length());
    }

    /**
     * 根据当前时间计算出当前滑动窗口的开始时间
     */
    protected long calculateWindowStart(long timeMillis) {
        // 利用求余运算原则，保证一秒内获取到的桶的开始时间是一致的
        // 1000 - 1000 % 1000 = 1000 - 0 = 1000
        // 1010 - 1010 % 1000 = 1010 - 10 = 1000
        // 2020 - 2020 % 1000 = 2020 - 20 = 2000
        return timeMillis - timeMillis % windowLengthInMs;
    }

    /**
     * 以提供的时间戳获取bucket。
     *
     * @param timeMillis 以毫秒为单位的有效时间戳
     * @return 如果时间有效，则在提供的时间戳处的当前bucket；如果时间无效，则为null
     */
    public WindowWrap<T> currentWindow(long timeMillis) {
        if (timeMillis < 0) {
            return null;
        }
        // 根据当前时间计算出当前时间属于那个滑动窗口的数组下标
        int idx = calculateTimeIdx(timeMillis);
        // 根据当前时间计算出当前滑动窗口的开始时间
        long windowStart = calculateWindowStart(timeMillis);

        /*
         * 根据下脚标在环形数组中获取滑动窗口（桶）
         *
         * (1) 如果桶不存在则创建新的桶，并通过CAS将新桶赋值到数组下标位。
         * (2) 如果获取到的桶不为空，并且桶的开始时间等于刚刚算出来的时间，那么返回当前获取到的桶。
         * (3) 如果获取到的桶不为空，并且桶的开始时间小于刚刚算出来的开始时间，那么说明这个桶是上一圈用过的桶，重置当前桶
         * (4) 如果获取到的桶不为空，并且桶的开始时间大于刚刚算出来的开始时间，理论上不应该出现这种情况。
         */
        while (true) {
            WindowWrap<T> old = array.get(idx);
            if (old == null) {
                /*
                 *     B0       B1      B2    NULL      B4
                 * ||_______|_______|_______|_______|_______||___
                 * 200     400     600     800     1000    1200  timestamp
                 *                             ^
                 *                          time=888
                 *            bucket is empty, 因此，创建新的和更新
                 *
                 * 如果旧的bucket不存在，那么我们在 {@code windowStart},
                 * 然后尝试通过CAS操作更新循环数组. 只有一根线程可以成功更新，而其他线程产生其时间片
                 */
                WindowWrap<T> window = new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
                if (array.compareAndSet(idx, null, window)) {
                    // 更新成功，返回创建的bucket.
                    return window;
                } else {
                    // 争用失败，线程将产生其时间片以等待可用的桶。
                    Thread.yield();
                }
            } else if (windowStart == old.windowStart()) {
                /*
                 *     B0       B1      B2     B3      B4
                 * ||_______|_______|_______|_______|_______||___
                 * 200     400     600     800     1000    1200  timestamp
                 *                             ^
                 *                          time=888
                 *            startTime of Bucket 3: 800
                 *
                 * 如果当前{@code windowStart}等于旧bucket的开始时间戳,
                 * 这意味着时间在bucket内，因此直接返回bucket.
                 */
                return old;
            } else if (windowStart > old.windowStart()) {
                /*
                 *   (old)
                 *             B0       B1      B2    NULL      B4
                 * |_______||_______|_______|_______|_______|_______||___
                 * ...    1200     1400    1600    1800    2000    2200  timestamp
                 *                              ^
                 *                           time=1676
                 *          startTime of Bucket 2: 400, 已弃用，应重置
                 *
                 * 如果旧bucket的开始时间戳晚于提供的时间，则表示该bucket已弃用。我们必须将bucket重置为当前{@code windowStart}.
                 * 请注意，重置和清理操作很难原子化,所以我们需要一个更新锁来保证bucket更新的正确性.
                 *
                 * 更新锁是有条件的（范围很小），只有当bucket不推荐使用时才会生效，所以在大多数情况下它不会导致性能损失。
                 */
                if (updateLock.tryLock()) {
                    try {
                        // 成功获得更新锁，现在我们重置桶。
                        return resetWindowTo(old, windowStart);
                    } finally {
                        updateLock.unlock();
                    }
                } else {
                    // 争用失败，线程将产生其时间片以等待可用的桶。
                    Thread.yield();
                }
            } else if (windowStart < old.windowStart()) {
                // 不应该通过这里，因为提供的时间已经落后了。
                return new WindowWrap<T>(windowLengthInMs, windowStart, newEmptyBucket(timeMillis));
            }
        }
    }

    /**
     * 检查某个bucket是否已弃用，这意味着该bucket已落后至少一整段时间。
     *
     * @param windowWrap 非空桶
     * @return 如果bucket已弃用，则为true；否则是错误的
     */
    public boolean isWindowDeprecated(WindowWrap<T> windowWrap) {
        return isWindowDeprecated(SystemClock.getNow(), windowWrap);
    }

    public boolean isWindowDeprecated(long time, WindowWrap<T> windowWrap) {
        return time - windowWrap.windowStart() > intervalInMs;
    }

    /**
     * 获取整个滑动窗口的有效桶列表。该列表将仅包含“valid”bucket。
     *
     * @return 整个滑动窗口的有效bucket列表。
     */
    public List<WindowWrap<T>> list() {
        return list(SystemClock.getNow());
    }

    public List<WindowWrap<T>> list(long validTime) {
        int size = array.length();
        List<WindowWrap<T>> result = new ArrayList<WindowWrap<T>>(size);

        for (int i = 0; i < size; i++) {
            WindowWrap<T> windowWrap = array.get(i);
            if (windowWrap == null || isWindowDeprecated(validTime, windowWrap)) {
                continue;
            }
            result.add(windowWrap);
        }

        return result;
    }

    /**
     * 获取整个滑动窗口的所有bucket，包括不推荐使用的bucket。
     *
     * @return 整个滑动窗口的所有bucket
     */
    public List<WindowWrap<T>> listAll() {
        int size = array.length();
        List<WindowWrap<T>> result = new ArrayList<WindowWrap<T>>(size);

        for (int i = 0; i < size; i++) {
            WindowWrap<T> windowWrap = array.get(i);
            if (windowWrap == null) {
                continue;
            }
            result.add(windowWrap);
        }

        return result;
    }

}
